package main

import (
	"os/exec"
	"sync"
	"sync/atomic"

	log "github.com/Sirupsen/logrus"
)

const (
	CONF_JOB_SCHEDULER = "job.scheduler"
	CONF_JOB_TYPE      = "job.type"
	CONF_JOB_ID        = "job.id"
	CONF_JOB_RUNSERVER = "job.runserver"
	CONF_JOB_COST      = "job.cost"
	CONF_ACTION_TYPE   = "dw.action.type"
	CONF_CALLBACK_URL  = "job.end.notification.url"

	DIST_FROM = "dist_from_path"
	DIST_TO   = "dist_to_path"

	JOB_TYPE_CMD         = "COMMAND"
	JOB_TYPE_DIST        = "DISTRIBUTION"
	JOB_TYPE_DIST_UPLOAD = "DISTRIBUTION_UPLOAD"

	FILE_OWNER = "file.owner"

	JOB_STATUS_RUNNING     = 1
	JOB_STATUS_KILLED      = 2
	JOB_STATUS_SUCCEED     = 3
	JOB_STATUS_FAILED      = 4
	JOB_STATUS_KILL_FAILED = 5
)

type JobConf map[string]string

func NewJobConf() JobConf {
	return make(map[string]string)
}

func (this JobConf) Get(key string) string {
	if value, ok := this[key]; ok {
		return value
	}
	return ""
}

func (this JobConf) Set(key, value string) {
	this[key] = value
}

func (this JobConf) Delete(key string) {
	delete(this, key)
}

type Properties map[string]string

func NewProperties() Properties {
	return make(map[string]string)
}

func (this Properties) Get(key string) string {
	if value, ok := this[key]; ok {
		return value
	}
	return ""
}

func (this Properties) Set(key, value string) {
	this[key] = value
}

func (this Properties) Delete(key string) {
	delete(this, key)
}

var sendIndex int32
var lastSendSuccessIndex int32

type JobResult struct {
	Id         string     `json:"id"`
	Status     int        `json:"status"`
	ErrDesc    string     `json:"errDesc"`
	Properties Properties `json:"properties"`
	RunServer  string     `json:"runServer"`
	JobConf    JobConf    `json:"jobConf"`

	SendIndex int32 `json:"sendIndex"`
}

func NewJobResult() *JobResult {
	atomic.AddInt32(&sendIndex, 1)
	res := &JobResult{
		SendIndex: sendIndex,
	}
	return res
}

type JobHandler struct {
	jobQueue    chan JobConf
	concurrency int
	wg          *sync.WaitGroup
	ctx         *Worker
	runningJobs *JobMap
	done        chan int // 标识任务已经执行完成
	logDone     chan int // 标识日志已经生成完成
	sendDone    chan int // 标识日志已经发送完成

	curJob         JobConf //只有一个job的时候，保存接收到的job信息
	logRespChan    chan bool
	unsendDoneChan chan bool //表示所有日志都已经被收到
}

type RunningJob struct {
	jobConf JobConf
	cmd     *exec.Cmd
}

type JobMap struct {
	m    map[string]*RunningJob
	lock sync.RWMutex
}

func NewJobMap() *JobMap {
	return &JobMap{
		m: make(map[string]*RunningJob),
	}
}

func (this *JobMap) Get(key string) *RunningJob {
	this.lock.RLock()
	defer this.lock.RUnlock()
	return this.m[key]
}

func (this *JobMap) Set(key string, job *RunningJob) {
	this.lock.Lock()
	defer this.lock.Unlock()
	this.m[key] = job
}

func (this *JobMap) Delete(key string) *RunningJob {
	this.lock.Lock()
	defer this.lock.Unlock()
	var j *RunningJob
	var ok bool
	if j, ok = this.m[key]; ok {
		delete(this.m, key)
	}
	return j
}

func (this *JobMap) Len() int {
	this.lock.RLock()
	defer this.lock.RUnlock()
	return len(this.m)
}

func (this *JobMap) Iter() <-chan *RunningJob {
	ch := make(chan *RunningJob)
	go func() {
		this.lock.RLock()
		for _, job := range this.m {
			ch <- job
		}
		this.lock.RUnlock()
		close(ch)
	}()
	return ch
}

func NewJobHandler(ctx *Worker, concurrency int) *JobHandler {
	if concurrency == 0 {
		concurrency = 1
	}
	handler := JobHandler{
		jobQueue:       make(chan JobConf, concurrency),
		concurrency:    concurrency,
		wg:             &sync.WaitGroup{},
		ctx:            ctx,
		runningJobs:    NewJobMap(),
		done:           make(chan int, 1),
		logRespChan:    make(chan bool, 1),
		unsendDoneChan: make(chan bool, 1),
		logDone:        make(chan int, 1),
		sendDone:       make(chan int, 1),
	}
	return &handler
}

func (this *JobHandler) HandleJob() {
	for i := 0; i < this.concurrency; i++ {
		this.wg.Add(1)
		go func() {
			this.handleJob()
			this.wg.Done()
		}()
	}
}

func (this *JobHandler) AddJob(job JobConf) {
	this.jobQueue <- job
	this.curJob = job
}

func (this *JobHandler) handleJob() {
	for jobConf := range this.jobQueue {
		log.Infof("handle job :%+v", jobConf)
		jobType := jobConf.Get(CONF_JOB_TYPE)
		jobId := jobConf.Get(CONF_JOB_ID)
		switch jobType {
		case JOB_TYPE_CMD:
			this.CmdJobExecute(jobConf)
		case JOB_TYPE_DIST:
		case JOB_TYPE_DIST_UPLOAD:
		default:
			log.Errorf("job handler: unsupported job type %s", jobType)
			jobResult := NewJobResult()
			jobResult.Id = jobId
			jobResult.Status = JOB_STATUS_FAILED
			jobResult.JobConf = jobConf
			jobResult.ErrDesc = "not support job type:" + jobType
			msg := &Message{
				Type: MT_JOB_RESULT,
				Body: jobResult,
			}
			this.ctx.FlushMessage(msg)
		}
	}
}

func (this *JobHandler) Stop() {
	log.Infof("stop handlers")
	close(this.jobQueue)
}

func (this *JobHandler) Wait() {
	this.wg.Wait()
}

func (this *JobHandler) waitUnsendMessageDone() {
	<-this.unsendDoneChan
}
